﻿using Basic.EventBus.Eventbus;
using Confluent.Kafka;
using Microsoft.Extensions.Logging;

namespace Basic.EventBus.EventBusKafka
{
    public class KafkaSubscriber : IEventSubscriber
    {
        private ILogger<KafkaProducer> Logger;
        private KafkaBusBrokerConfig Config;
        private List<KafkaConsumerRecord> RecordList;
        /// <summary>
        /// 消费者运行
        /// </summary>
        private bool Running = false;

        public KafkaSubscriber(ILogger<KafkaProducer> logger, KafkaBusBrokerConfig options)
        {
            Logger = logger;
            Config = options;
            RecordList = new List<KafkaConsumerRecord>(8);
        }

        public void AddSubscriber(ISubscriberHandler subscriberInfo)
        {
            var record = new KafkaConsumerRecord(Logger, subscriberInfo, Config);
            record.Start();
            RecordList.Add(record);
        }

        public void AddSubscriber(ISubscriberHandler[] subscriberInfos)
        {
            if (subscriberInfos?.Length > 0)
                foreach (var item in subscriberInfos)
                    AddSubscriber(item);
        }

        #region 消费逻辑

        internal class KafkaConsumerRecord
        {
            private ILogger Logger;
            private bool AutoAck;
            private SemaphoreSlim ConcurrentLock;
            private KafkaBusBrokerConfig Config;

            public IConsumer<string, byte[]> Consumer { get; private set; }
            public ISubscriberHandler Subscriber { get; private set; }
            public DateTime CreateTime { get; private set; }

            public KafkaConsumerRecord(ILogger logger, ISubscriberHandler handler, KafkaBusBrokerConfig config)
            {
                Logger = logger;
                Subscriber = handler;
                Config = config;
                AutoAck = config.AutoAck;

                int initNum = 8;
                if (initNum > config.Qos) initNum = config.Qos;
                ConcurrentLock = new SemaphoreSlim(initNum, config.Qos);
            }

            private IConsumer<string, byte[]> CreateConsumerHanlder()
            {
                ConsumerConfig conf = new ConsumerConfig()
                {
                    BootstrapServers = Config.Servers,
                    GroupId = Subscriber.Attribute.Group,
                    AutoOffsetReset = AutoOffsetReset.Earliest,
                    EnableAutoCommit = AutoAck
                };
                var builder = new ConsumerBuilder<string, byte[]>(conf);
                return builder.Build();
            }

            public void Start()
            {
                Consumer = CreateConsumerHanlder();
                Consumer.Subscribe(Subscriber.Attribute.Topic);

                _ = Task.Factory.StartNew(async () =>
                {
                    while (Subscriber.TokenSource.Token.CanBeCanceled)
                    {
                        await ReceiveMessage();
                    }
                    Consumer.Dispose();
                    Consumer = null;
                }, Subscriber.TokenSource.Token);
            }

            private async Task ReceiveMessage()
            {
                try
                {
                    await ConcurrentLock.WaitAsync();

                    var result = Consumer.Consume();
                    ThreadPool.QueueUserWorkItem(async obj =>
                    {
                        await Subscriber.HandleMessage(obj as byte[]);
                        ConcurrentLock.Release();
                    }, result.Message.Value);

                    if (!AutoAck) Consumer.Commit(result);
                }
                catch (Exception ex)
                {
                    Logger.LogError(ex, "KafkaConsumerRecord.ReceiveMessage()->Error:");
                }
            }

        }

        #endregion


        #region Dispose
        private bool disposedValue;

        protected virtual void Dispose(bool disposing)
        {
            if (!disposedValue)
            {
                if (disposing)
                {
                    // TODO: 释放托管状态(托管对象)
                }
                // TODO: 将大型字段设置为 null
                disposedValue = true;
            }
        }

        public void Dispose()
        {
            // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
            Dispose(disposing: true);
            GC.SuppressFinalize(this);
        }
        #endregion

    }
}